package cn.sciento.transfer.app.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import cn.sciento.core.exception.CommonException;import java.sql.Connection;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;
import liquibase.database.Database;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import cn.sciento.core.base.BaseConstants;
import cn.sciento.transfer.app.service.DbMigrationLogService;
import cn.sciento.transfer.domain.entity.ConsDbConfig;
import cn.sciento.transfer.domain.entity.DbMigrationLog;
import cn.sciento.transfer.domain.entity.ProducerConfig;
import cn.sciento.transfer.domain.repository.ConsDbConfigRepository;
import cn.sciento.transfer.domain.repository.DbMigrationLogRepository;
import cn.sciento.transfer.domain.repository.ProducerConfigRepository;
import cn.sciento.transfer.infra.datasource.utils.AsyncExecutionTemplate;
import cn.sciento.transfer.infra.datasource.utils.TransferDataSourceTemplate;
import cn.sciento.transfer.infra.liquibase.parser.data.DatabaseChangeLogFileData;
import cn.sciento.transfer.infra.migration.DBMigrationUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

@Service
public class DbMigrationLogServiceImpl implements DbMigrationLogService {
  private static final Logger LOGGER = LoggerFactory.getLogger(DbMigrationLogServiceImpl.class);
  
  @Autowired
  private ProducerConfigRepository producerConfigRepository;
  
  @Autowired
  private ConsDbConfigRepository consDbConfigRepository;
  
  @Autowired
  private TransferDataSourceTemplate transferDataSourceTemplate;
  
  @Autowired
  private DbMigrationLogRepository dbMigrationLogRepository;
  
  @Autowired
  private SqlSessionFactory sqlSessionFactory;
  
  public List<DbMigrationLog> probeDbMigrations() {
    LOGGER.debug("===================start probe db migrations===================");
    List<ProducerConfig> producerConfigList = fetchEffectiveConfig();
    if (CollectionUtils.isEmpty(producerConfigList))
      return Collections.emptyList(); 
    List<DbMigrationLog> dbMigrationLogList = new ArrayList<>(16);
    producerConfigList.forEach(producerConfig -> {
          List<DbMigrationLog> internalDbMigrationLogList = internalProbeDbMigrations(producerConfig);
          dbMigrationLogList.addAll(internalDbMigrationLogList);
        });
    if (!CollectionUtils.isEmpty(dbMigrationLogList))
      this.dbMigrationLogRepository.batchInsertSelective(dbMigrationLogList); 
    LOGGER.debug("===================exit probe db migrations===================");
    return dbMigrationLogList;
  }
  
  public String doExecuteDbMigration(Long dbMigrationLogId) {
    DbMigrationLog dbMigrationLog = (DbMigrationLog)this.dbMigrationLogRepository.selectByPrimaryKey(dbMigrationLogId);
    Assert.notNull(dbMigrationLog, "db migration cannot be null.");
    if (!"P".equals(dbMigrationLog.getProcessStatus()) && 
      !"E".equals(dbMigrationLog.getProcessStatus()))
      throw new CommonException("invalid process status.");
    try {
      String updateLogDetail = (String)this.transferDataSourceTemplate.doExecuteInConnection(dbMigrationLog
          .getTargetDs(), dbMigrationLog
          .getTargetService(), dbMigrationLog
          .getTargetDb(), targetConnection -> {
            try {
              Database targetDatabase = DBMigrationUtils.buildDatabase(targetConnection);
              return DBMigrationUtils.updateDatabase((DatabaseChangeLogFileData)(new ObjectMapper()).readValue(dbMigrationLog.getMessages(), DatabaseChangeLogFileData.class), targetDatabase);
            } catch (Exception ex) {
              throw new CommonException(ex);
            } 
          });
      dbMigrationLog.setProcessStatus("S");
      dbMigrationLog.setProcessMsg(updateLogDetail);
    } catch (Exception ex) {
      dbMigrationLog.setProcessStatus("E");
      dbMigrationLog.setProcessMsg(ExceptionUtils.getStackTrace(ex));
    } 
    dbMigrationLog.setProcessTime(new Date());
    this.dbMigrationLogRepository.updateOptional(dbMigrationLog, new String[] { "processStatus", "processMsg", "processTime" });
    internalResetHistoryPendingData(dbMigrationLog);
    LOGGER.debug("====================>exit execute change log<====================");
    return dbMigrationLog.getProcessMsg();
  }
  
  private List<ProducerConfig> fetchEffectiveConfig() {
    ProducerConfig sampleProducerConfig = new ProducerConfig();
    sampleProducerConfig.setEnabledFlag(BaseConstants.Flag.YES);
    List<ProducerConfig> producerConfigList = this.producerConfigRepository.select(sampleProducerConfig);
    if (CollectionUtils.isEmpty(producerConfigList))
      return Collections.emptyList(); 
    producerConfigList.forEach(producerConfig -> {
          ConsDbConfig sampleConsDbConfig = new ConsDbConfig();
          sampleConsDbConfig.setProducerConfigId(producerConfig.getProducerConfigId());
          sampleConsDbConfig.setEnabledFlag(BaseConstants.Flag.YES);
          List<ConsDbConfig> consDbConfigList = this.consDbConfigRepository.select(sampleConsDbConfig);
          if (CollectionUtils.isEmpty(consDbConfigList))
            return; 
          producerConfig.setConsDbConfigs(consDbConfigList);
        });
    return producerConfigList;
  }
  
  private List<DbMigrationLog> internalProbeDbMigrations(ProducerConfig producerConfig) {
    List<DbMigrationLog> dbMigrationLogList = new ArrayList<>(16);
    List<ConsDbConfig> consDbConfigList = producerConfig.getConsDbConfigs();
    if (CollectionUtils.isEmpty(consDbConfigList)) {
      return Collections.emptyList();
    } else {
      this.transferDataSourceTemplate.doExecuteInConnection(producerConfig.getInitDsCode(), producerConfig.getServiceCode(), producerConfig.getInitDbCode(), (sourceConnection) -> {
        try {
          Database sourceDatabase = DBMigrationUtils.buildDatabase(sourceConnection);
          consDbConfigList.forEach((consDbConfig) -> {
            DbMigrationLog dbMigrationLog = this.internalProbeDbMigration(producerConfig, consDbConfig, sourceDatabase);
            if (dbMigrationLog != null) {
              dbMigrationLogList.add(dbMigrationLog);
            }

          });
          return null;
        } catch (Exception var6) {
          LOGGER.error("internalProbeDbMigrations with exception: ");
          throw new CommonException(var6);
        }
      });
      return dbMigrationLogList;
    }
  }

  private DbMigrationLog internalProbeDbMigration(ProducerConfig producerConfig, ConsDbConfig consDbConfig, Database sourceDatabase) {
    DatabaseChangeLogFileData databaseChangeLogFileData = (DatabaseChangeLogFileData)AsyncExecutionTemplate.asyncDoExecuteAndAwait(AsyncExecutionTemplate.findAsyncTaskExecutor("transferTargetTaskExecutor"), () -> {
      DatabaseChangeLogFileData innerDatabaseChangeLogFileData = (DatabaseChangeLogFileData)this.transferDataSourceTemplate.doExecuteInConnection(consDbConfig.getConsumerDs(), consDbConfig.getConsumerService(), consDbConfig.getConsumerDb(), (targetConnection) -> {
        try {
          Database targetDatabase = DBMigrationUtils.buildDatabase(targetConnection);
          return DBMigrationUtils.compareDatabase(sourceDatabase, targetDatabase, producerConfig.getTableName(), consDbConfig.getConsumerTable());
        } catch (Exception var5) {
          throw new CommonException(var5);
        }
      });
      return innerDatabaseChangeLogFileData;
    });
    if (databaseChangeLogFileData != null && !StringUtils.isEmpty(databaseChangeLogFileData.getFileContent())) {
      try {
        DBMigrationUtils.stripRemarkChanges(databaseChangeLogFileData);
      } catch (Exception var8) {
        LOGGER.error("strip remark changes from change log with error, ignore: ", var8);
      }

      if (StringUtils.isEmpty(databaseChangeLogFileData.getFileContent())) {
        LOGGER.debug("end diff uuid [{}], no diff except remark changes.");
        return null;
      } else {
        DbMigrationLog dbMigrationLog = new DbMigrationLog();
        dbMigrationLog.setSourceService(producerConfig.getServiceCode());
        dbMigrationLog.setSourceDsId(producerConfig.getInitDsId());
        dbMigrationLog.setSourceDs(producerConfig.getInitDsCode());
        dbMigrationLog.setSourceDb(producerConfig.getInitDbCode());
        dbMigrationLog.setSourceTable(producerConfig.getTableName());
        dbMigrationLog.setTargetService(consDbConfig.getConsumerService());
        dbMigrationLog.setTargetDsId(consDbConfig.getConsumerDsId());
        dbMigrationLog.setTargetDs(consDbConfig.getConsumerDs());
        dbMigrationLog.setTargetDb(consDbConfig.getConsumerDb());
        dbMigrationLog.setTargetTable(consDbConfig.getConsumerTable());

        try {
          dbMigrationLog.setMessages((new ObjectMapper()).writeValueAsString(databaseChangeLogFileData));
        } catch (Exception var7) {
          LOGGER.error("parse change log content with object mapper with error: ");
          throw new CommonException(var7);
        }

        dbMigrationLog.setProcessStatus("P");
        return dbMigrationLog;
      }
    } else {
      LOGGER.debug("end diff uuid [{}], no diff.");
      return null;
    }
  }
  
  private void internalResetHistoryPendingData(DbMigrationLog dbMigrationLog) {
    DbMigrationLog sampleDbMigrationLog = new DbMigrationLog();
    sampleDbMigrationLog.setSourceService(dbMigrationLog.getSourceService());
    sampleDbMigrationLog.setSourceDsId(dbMigrationLog.getSourceDsId());
    sampleDbMigrationLog.setSourceDs(dbMigrationLog.getSourceDs());
    sampleDbMigrationLog.setSourceDb(dbMigrationLog.getSourceDb());
    sampleDbMigrationLog.setSourceTable(dbMigrationLog.getSourceTable());
    sampleDbMigrationLog.setTargetService(dbMigrationLog.getTargetService());
    sampleDbMigrationLog.setTargetDsId(dbMigrationLog.getTargetDsId());
    sampleDbMigrationLog.setTargetDs(dbMigrationLog.getTargetDs());
    sampleDbMigrationLog.setTargetDb(dbMigrationLog.getTargetDb());
    sampleDbMigrationLog.setTargetTable(dbMigrationLog.getTargetTable());
    sampleDbMigrationLog.setProcessStatus("P");
    List<DbMigrationLog> pendingDbMigrationLogList = this.dbMigrationLogRepository.select(sampleDbMigrationLog);
    sampleDbMigrationLog.setProcessStatus("E");
    List<DbMigrationLog> errorDbMigrationLogList = this.dbMigrationLogRepository.select(sampleDbMigrationLog);
    List<DbMigrationLog> all = new ArrayList<>(2);
    all.addAll(pendingDbMigrationLogList);
    all.addAll(errorDbMigrationLogList);
    if (!CollectionUtils.isEmpty(all)) {
      List<DbMigrationLog> resetList = (List<DbMigrationLog>)all.parallelStream().filter(item -> (Long.compare(dbMigrationLog.getDbMigrationLogId().longValue(), item.getDbMigrationLogId().longValue()) > 0)).collect(Collectors.toList());
      resetList.forEach(item -> {
            item.setProcessStatus("N");
            item.setProcessTime(new Date());
            item.setProcessMsg(String.format("ahead reacord %s has been executed.", new Object[] { dbMigrationLog.getDbMigrationLogId() }));
          });
      this.dbMigrationLogRepository.batchUpdateOptional(resetList, new String[] { "processStatus", "processMsg", "processTime" });
    } 
  }
  
  private void testSqlSessionFactory() {
    SqlSession sqlSession = this.sqlSessionFactory.openSession();
    Connection connection = null;
    try {
      connection = sqlSession.getConnection();
      LOGGER.debug("====================schema: {}, catalog: {}====================", connection.getSchema(), connection.getCatalog());
    } catch (Exception ex) {
      sqlSession.close();
    } finally {
      sqlSession.close();
    } 
  }
}
